package com.lagou.client;

import com.lagou.common.constant.Constants;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * @ClassName: ServiceDiscovery
 * @Description: TODO
 * @Author: lixing
 * @Date: 2020-04-03 20:11
 * @Version: 0.0.1
 */
public class ServiceDiscovery implements Watcher {
    private CountDownLatch latch = new CountDownLatch(1);

    private volatile List<String> serviceAddressList = new ArrayList<>();

    private ZooKeeper zk = null;

    private String registryAddress; // 注册中心的地址

    public ServiceDiscovery(String registryAddress) {
        this.registryAddress = registryAddress;

        zk = connectServer();
        if (zk != null) {
            watchNode(zk);
        }
    }

    /**
     * 通过服务发现，获取服务提供方的地址
     *
     * @return
     */
    public String discover() {
        String data = null;
        int size = serviceAddressList.size();
        if (size > 0) {
            if (size == 1) {  //只有一个服务提供方
                data = serviceAddressList.get(0);
                System.out.println("只有一个服务提供者，地址为 : {" + data + "}");
            } else {          //使用随机分配法。简单的负载均衡法
                //data = serviceAddressList.get(ThreadLocalRandom.current().nextInt(size));
                data = loadBanlanceServer(serviceAddressList);
                System.out.println("选中的调用服务提供者地址为 : {" + data + "}");
            }
        }
        return data;
    }

    /**
     * 负载均衡获取要调用的服务
     * 1. Zookeeper记录每个服务端的最后一次响应时间，有效时间为5秒，5s内如果该服务端没有新的请求，响应时间清零或失效
     * 2. 当客户端发起调用，每次都选择最后一次响应时间短的服务端进行服务调用，如果时间一致，随机选取一个服务端进行调用，从而实现负载均衡
     *
     * @param serviceAddressList
     * @return
     */
    private String loadBanlanceServer(List<String> serviceAddressList) {
        String serverAddress = "";
        // 记录最后一次的响应时间
        Map responseTimeMap = new HashMap();
        // 记录服务调用方ip
        List iPList = new ArrayList();
        // 记录服务调用方花费时间
        List costList = new ArrayList();
        for (int i = 0; i < serviceAddressList.size(); i++) {
            String data = serviceAddressList.get(i);
            String[] strings = data.split(";");
            if (strings.length > 1) {
                responseTimeMap.put(strings[0], strings[1]);
                iPList.add(strings[0]);
                costList.add(Long.parseLong(strings[2]));
            } else {
                serverAddress = strings[0];
            }
        }
        if (!responseTimeMap.isEmpty()) {
            responseTimeMap.forEach((key, value) -> {
                System.out.println("当前ip为:" + key + ";最后一次响应时间为：" + value);
                if (TimeUnit.SECONDS.toSeconds(System.currentTimeMillis() - Long.parseLong((String) value)) > 5) {
                    // Zookeeper记录每个服务端的最后一次响应时间，有效时间为5秒，5s内如果该服务端没有新的请求，响应时间清零或失效
                    System.out.println("该ip地址的响应时间失效：" + key);
                    try {
                        zk.setData(Constants.ZK_DATA_PATH, (key + ";0").getBytes(), -1);
                    } catch (KeeperException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {

                }
            });
        }
        // 每次都选择最后一次响应时间短的服务端进行服务调用，如果时间一致，随机选取一个服务端进行调用，从而实现负载均衡
        if (!costList.isEmpty()) {
            // 如果响应时间一样，maxIndex会随机返回一个值
            int maxIndex = costList.indexOf(Collections.min(costList));
            serverAddress = (String) iPList.get(maxIndex);
        }
        System.out.println("负载均衡后调用的服务地址为：" + serverAddress);
        return serverAddress;
    }

    /**
     * 连接 zookeeper
     *
     * @return
     */
    private ZooKeeper connectServer() {

        ZooKeeper zk = null;
        try {
            zk = new ZooKeeper(registryAddress, Constants.ZK_SESSION_TIMEOUT, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    System.out.println(event.getType() + "---" + event.getPath());
                    if (event.getState() == Watcher.Event.KeeperState.SyncConnected) {
                        latch.countDown();
                    }
                }
            });
            latch.await();
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
        return zk;
    }

    /**
     * 获取服务地址列表
     *
     * @param zk
     */
    private void watchNode(final ZooKeeper zk) {

        try {
            //获取子节点列表
            List<String> nodeList = zk.getChildren(Constants.ZK_REGISTRY_PATH, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {
                        //发生子节点变化时再次调用此方法更新服务地址
                        watchNode(zk);
                        try {
                            RpcConsumer.initClient();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            });
            List<String> dataList = new ArrayList<>();
            for (String node : nodeList) {
                byte[] bytes = zk.getData(Constants.ZK_REGISTRY_PATH + "/" + node, false, null);
                dataList.add(new String(bytes));
            }
            System.out.println("服务地址信息为: " + dataList);
            this.serviceAddressList = dataList;
        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        Event.KeeperState zkState = watchedEvent.getState();
        String path = watchedEvent.getPath();
        Event.EventType eventType = watchedEvent.getType();

        if (zkState == Event.KeeperState.SyncConnected) {
            if (eventType == Event.EventType.None) {
                //连接建立事件
                System.out.println("建立zookeeper连接成功!");
                latch.countDown();
            } else if (eventType == Event.EventType.NodeCreated) {
                //节点创建事件
                System.out.println("节点" + path + "被创建");
            } else if (eventType == Event.EventType.NodeDataChanged) {
                //节点修改事件
                System.out.println("节点" + path + "被修改");

            } else if (eventType == Event.EventType.NodeDeleted) {
                //节点删除事件
                System.out.println("节点" + path + "被删除");
            }
        }
    }
}
